package com.xueyuan.wata.daph.node.flink117.string.general.transformer

import com.xueyuan.wata.daph.flink117.api.node.string.StringMultipleTransformer
import com.xueyuan.wata.daph.node.flink117.string.general.{GeneralNode, GeneralSQLConfig}

class GeneralSQL extends StringMultipleTransformer with GeneralNode {
  override def transform(lineToDS: Map[String, String]): String = {
    val config = nodeConfig.asInstanceOf[GeneralSQLConfig]
    val resultTableName = config.resultTableName
    val sqlStatement = config.sqlStatement

    // 执行查询语句，创建视图到默认catalog
    createTemporaryView(resultTableName, sqlStatement)

    // 传递表名
    resultTableName
  }

  override def getNodeConfigClass = classOf[GeneralSQLConfig]
}
